#include "TcpClient.h"
#include <arpa/inet.h>
#include <netinet/tcp.h>
#include <unistd.h>

#include <cstring>
#include <iostream>
#include <thread>
#include <vector>

namespace BORNS {

TcpClient::TcpClient(bool selectTimeout,
                     TcpProtocol::Type p,
                     TcpProtocol::Device dev) :
    socket_fd_(-1),
    state_(SocketState::Invalid),
    uniqueId_(0),
    proType_(p),
    dev_(dev),
    hostIp_("127.0.0.1"),
    port_(-1),
    selectTimeoutError_(selectTimeout),
    rcvThreadRunning_(false),
    pauseRcvThread_(false),
    rcvThreadIsPause_(false)
{

}
TcpClient::~TcpClient()
{
    rcvThreadRunning_.store(false);
    Close();
}

void TcpClient::setOptions(int socket_fd)
{
    int flag = 1;
    setsockopt(socket_fd, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(int));
#if defined(__linux) || defined(linux) || defined(__linux__)
    setsockopt(socket_fd, IPPROTO_TCP, TCP_QUICKACK, &flag, sizeof(int));
#endif
    if (recv_timeout_ != nullptr)
    {
        setsockopt(socket_fd, SOL_SOCKET, SO_RCVTIMEO, recv_timeout_.get(), sizeof(timeval));
    }
}

bool TcpClient::Connect(const std::string& host, int port)
{
    if (state_ == SocketState::Connected)
        return true;

    hostIp_ = host;
    port_ = port;
    struct sockaddr_in serverAddr;
    // 创建socket
    this->socket_fd_ = ::socket(AF_INET, SOCK_STREAM, 0);
    if(socket_fd_ < 0){
        std::cout<< "TCP::Failed to create socket" <<std::endl;
        return false;
    }
    serverAddr.sin_family = AF_INET;
    serverAddr.sin_port = htons(port);
    serverAddr.sin_addr.s_addr = inet_addr(host.c_str());
    // 连接
    int ret = ::connect(socket_fd_, (struct sockaddr *)&serverAddr,
                        sizeof(struct sockaddr));
    if (ret < 0)
    {
        state_ = SocketState::Invalid;
        std::cerr << "TCP::Connection setup failed for " << host.c_str() << ":" << port << std::endl;
        return false;
    }
    else
    {
        setOptions(socket_fd_);
        state_ = SocketState::Connected;
        if(proType_ == TcpProtocol::Type::HeaderData){
            // 链接建立成功后，发送this client所属设备种类给server端
            this->Write(reinterpret_cast<uint8_t *>(&dev_), sizeof (dev_));
        }
    }
    return true;
}

void TcpClient::Disconnect()
{
    if(rcvThreadRunning_.load()){
        this->PauseReceiveThread();// 先暂停接收线程
    }
    this->Close();// 清除状态并关闭套解字
}

bool TcpClient::Reconnect()
{
    this->Disconnect();
    if(this->Connect(hostIp_, port_)){
//        this->EnableReceiveThread();// 激活接收线程
        return true;
    }else{
        return false;
    }
}

void TcpClient::ReconnectBlock()
{
    this->Close();
    for(;;){
        std::this_thread::sleep_for(std::chrono::seconds(1));
        if(this->Connect(hostIp_, port_)){
            break;
        }
    }
}

bool TcpClient::SetSocketFD(int socket_fd)
{
    if (state_ == SocketState::Connected)
        return false;
    socket_fd_ = socket_fd;
    state_ = SocketState::Connected;
    return true;
}

void TcpClient::Close()
{
    if (socket_fd_ >= 0)
    {
        state_ = SocketState::Closed;
        ::close(socket_fd_);
        socket_fd_ = -1;
    }
}

void TcpClient::StopReceiveThread()
{
    this->rcvThreadRunning_.store(false);
}

void TcpClient::PauseReceiveThread()
{
    this->pauseRcvThread_.store(true);
    for(;;){
        if(this->rcvThreadIsPause_.load()){
            break;
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }
}

void TcpClient::EnableReceiveThread()
{
    this->pauseRcvThread_.store(false);
}

std::string TcpClient::GetIP() const
{
    return this->hostIp_;
}

bool TcpClient::Read(char* character)
{
    size_t read_chars;
    // It's inefficient, but in our case we read very small messages
    // and the overhead connected with reading character by character is
    // negligible - adding buffering would complicate the code needlessly.
    return Read((uint8_t*)character, 1, read_chars);
}

bool TcpClient::Read(uint8_t* buf, size_t buf_len, size_t& read)
{
    read = 0;

    if (state_ != SocketState::Connected)
        return false;

    ssize_t res = ::recv(socket_fd_, buf, buf_len, 0);

    if (res == 0)
    {
        state_ = SocketState::Disconnected;
        return false;
    }
    else if (res < 0)
        return false;

    read = static_cast<size_t>(res);
    return true;
}

bool TcpClient::Read(std::vector<uint8_t> &buf)
{
    if (state_ != SocketState::Connected)
        return false;

    size_t readCount = 0;
    size_t remaining = buf.size();
    while(readCount < buf.size()){
        ssize_t res = ::recv(socket_fd_, buf.data() + readCount, remaining, 0);

        if (res == 0)
        {
            state_ = SocketState::Disconnected;
            return false;
        }
        else if (res < 0)
            return false;
        readCount += res;
        remaining -= res;
    }
    return true;
}

bool TcpClient::Read(std::string &str)
{
    std::vector<uint8_t> buf;
    if(this->Read(buf)){
        str.assign(buf.begin(), buf.end());
        return true;
    }
    return false;
}

bool TcpClient::Write(const uint8_t *buf, size_t buf_len)
{
    size_t written = 0;
    if(proType_ == TcpProtocol::Type::HeaderData){
        uint8_t realBuf[TcpProtocol::maxWriteDataSize];
        size_t size = 0;
        TcpProtocol::PackBuffer(buf, buf_len, realBuf, size);
        if(this->Write(realBuf, size, written)){
            return true;
        }
    }else{
        if(this->Write(buf, buf_len, written)){
            return true;
        }
    }
    return false;
}

bool TcpClient::Write(const std::vector<uint8_t> &buf)
{
    return this->Write(buf.data(), buf.size());
}

bool TcpClient::Write(const void *buf, size_t buf_len)
{
    std::vector<uint8_t> realBuf(buf_len);
    memcpy(realBuf.data(), buf, buf_len);
    return this->Write(realBuf);
}

bool TcpClient::Write(const std::string &str)
{
    std::vector<uint8_t> realBuf(str.size());
    memcpy(realBuf.data(), str.data(), str.size());
    return this->Write(realBuf);
}

bool TcpClient::Write(const uint8_t* buf, size_t buf_len, size_t& written)
{
    written = 0;

    if (state_ != SocketState::Connected)
    {
        std::cerr << "Attempt to write on a non-connected socket" << std::endl;
        return false;
    }

    size_t remaining = buf_len;

    // handle partial sends
    while (written < buf_len)
    {
        ssize_t sent = ::send(socket_fd_, buf + written, remaining, 0);

        if (sent <= 0)
        {
            std::cerr << "Sending data through socket failed." << std::endl;
            return false;
        }

        written += sent;
        remaining -= sent;
    }
    return true;
}

void TcpClient::SetReceiveTimeout(const timeval& timeout)
{
    recv_timeout_.reset(new timeval(timeout));

    if (state_ == SocketState::Connected)
    {
        setOptions(socket_fd_);
    }
}

void TcpClient::RcvThread()
{
    struct timeval timeout;// select超时时间
    fd_set readfds;
    std::vector<uint8_t> readBuffer;// 读取数据的总缓存
    uint8_t buf[TcpProtocol::maxReadDataSize];// 每次读取的数据缓存
    while(rcvThreadRunning_){
        if(pauseRcvThread_){
            rcvThreadIsPause_.store(true);
            std::this_thread::sleep_for(std::chrono::seconds(1));
            continue;
        }
        rcvThreadIsPause_.store(false);
        bool error{false};
        FD_ZERO(&readfds);
        FD_SET(socket_fd_, &readfds);
        timeout.tv_sec = 0;  // do this each loop as selects modifies timeout
        timeout.tv_usec = 500000;  // timeout of 0.5 sec
        int selectRet = select(socket_fd_ + 1, &readfds, nullptr, nullptr, &timeout);
        // 文件描述符对应的IO有数据
        if(selectRet == 1){
            // 读取数据到buffer
            memset(buf, 0, TcpProtocol::maxReadDataSize);
            size_t bytesRead{0};
            // 读取成功
            if(this->Read(buf, TcpProtocol::maxReadDataSize, bytesRead)){
                if(readBuffer.size() >= TcpProtocol::maxReadBufferSize){
                    std::string error_str("TcpClient::RcvThread: readbuffer size overflow");
                    printf("%s\n", error_str.c_str());
                    throw std::runtime_error(error_str);
                }
                readBuffer.insert(readBuffer.cend(), buf, buf + bytesRead);
                if(proType_ == TcpProtocol::Type::HeaderData){
                    // 拆包并解析数据
                    Unpack(readBuffer);
                }else if(parseFunc_){
                    parseFunc_(readBuffer);
                }
            }else{// 读取错误
                if(state_ == SocketState::Disconnected){
                    error = true;
                    std::cout<<"TCP READ ERROR:disconnect: "<<hostIp_<<":"<<port_<<std::endl;
                }
            }
        }else if(selectRet == 0 && selectTimeoutError_.load()){
            static int selectFailedCount{0};//select超时
            ++selectFailedCount;
            if(selectFailedCount > 3){
                selectFailedCount = 0;
                error = true;
                std::cout<<"TCP SELECT TIMEOUT ERROR:disconnect: "<<hostIp_<<":"<<port_<<std::endl;
            }
        }else{// select异常
            continue;
        }
        // 清除状态，暂停该线程运行
        if(error){
            this->Close();
            pauseRcvThread_.store(true);
            // 复位状态
            memset(buf, 0, TcpProtocol::maxReadDataSize);
            readBuffer.clear();
        }
    }
}

size_t TcpClient::RcvId()
{
    struct timeval timeout;// select超时时间
    fd_set readfds;
    std::vector<uint8_t> readBuffer;// 读取数据的总缓存
    uint8_t buf[256];// 每次读取的数据缓存
    while(state_ == SocketState::Connected){
        FD_ZERO(&readfds);
        FD_SET(socket_fd_, &readfds);
        timeout.tv_sec = 1;  // do this each loop as selects modifies timeout
        timeout.tv_usec = 0;
        int selectRet = select(socket_fd_ + 1, &readfds, nullptr, nullptr, &timeout);
        // 文件描述符对应的IO有数据
        if(selectRet == 1){
            // 读取数据到buffer
            memset(buf, 0, 256);
            size_t bytesRead{0};
            // 读取成功
            if(this->Read(buf, 256, bytesRead)){
                readBuffer.insert(readBuffer.cend(), buf, buf + bytesRead);
                // 拆包并解析数据
                size_t id = 0;
                if(UnpackId(readBuffer, id)){
                    return id;
                }
            }else{// 读取错误

            }
        }
    }
    return 0;
}

//void TcpClient::SetConnectStatusChangedCallback(std::function<void (bool status)> statusChangedFunc)
//{
//    this->connectStatusChangedFunc_ = statusChangedFunc;
//}

void TcpClient::Unpack(std::vector<uint8_t> &bytes)
{
    while(bytes.size() >= TcpProtocol::PACKAGE_SIZE){
        // 获取数据包的包头信息即包的大小
        size_t packetSize = 0;
        memcpy(&packetSize, bytes.data(), TcpProtocol::PACKAGE_SIZE);
        if (bytes.size() >= packetSize && packetSize > 0) {
            // 删除包头信息
            bytes.erase(bytes.cbegin(), bytes.cbegin() + TcpProtocol::PACKAGE_SIZE);
            // 解析数据包
            if(parseFunc_ && !bytes.empty())
                parseFunc_(bytes);
            // 删除已解析的数据包
            bytes.erase(bytes.cbegin(), bytes.cbegin() + packetSize - TcpProtocol::PACKAGE_SIZE);
        } else {
            break;
        }
    }
}

bool TcpClient::UnpackId(std::vector<uint8_t> &bytes, size_t &id)
{
    if(bytes.size() >= TcpProtocol::PACKAGE_SIZE){
        // 获取UR数据包的包头信息即包的大小
        memcpy(&id, bytes.data(), TcpProtocol::PACKAGE_SIZE);
        return true;
    }
    return false;
}

bool TcpClient::StartReceiveThread(std::function<void(std::vector<uint8_t> &bytes)> parseFunc)
{
    this->parseFunc_ = parseFunc;
    if(!rcvThreadRunning_.load()){
        rcvThreadRunning_.store(true);
        std::thread(&TcpClient::RcvThread, this).detach();
    }
    this->EnableReceiveThread();
    return true;
}

}// namespace BORNS
